{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "87175262",
   "metadata": {},
   "source": [
    "## 易错点\n",
    "不是rdd也可以map\n",
    "transformedData = transformedData.map(lambda row: LabeledPoint(row[0],[row[1]]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d259b46e",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "sc=SparkContext('local','test')\n",
    "spark=SparkSession(sc)\n",
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6f728b88",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import Row\n",
    "houses_data = spark.read.format('org.apache.spark.sql.execution.datasources.csv.CSVFileFormat')\\\n",
    ".option('header','true')\\\n",
    ".option('inferSchema','true')\\\n",
    ".load('data/houses_data.csv')\n",
    "rdd = sc.textFile('data/houses_data.csv')\n",
    "rdd.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8c1e3c69",
   "metadata": {},
   "source": [
    "1.对数据集的每一行用逗号进行分隔。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9546999c",
   "metadata": {},
   "outputs": [],
   "source": [
    "#由考生填写\n",
    "rdd = rdd.map(lambda line: line.split(\",\"))\n",
    "#由考生填写\n",
    "rdd.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "74b78783",
   "metadata": {},
   "source": [
    "2.使用“fiter”删除包含标题的行。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "19534c73",
   "metadata": {},
   "outputs": [],
   "source": [
    "header = rdd.first()\n",
    "#由考生填写\n",
    "rdd = rdd.filter(lambda line: line != header)\n",
    "#由考生填写\n",
    "rdd.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "309d21b0",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = rdd.map(lambda line: Row(street = line[0], city = line[1], zip=line[2], beds=line[4], baths=line[5],sqft=line[6], price=line[9])).toDF()\n",
    "df.show()\n",
    "df.toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5036f213",
   "metadata": {},
   "source": [
    "3.df按照“beds”字段分组，并计算每个分组中记录的数量，并显示结果"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "32265f15",
   "metadata": {},
   "outputs": [],
   "source": [
    "# func:`groupby` is an alias for :func:`groupBy`.\n",
    "#由考生填写 \n",
    "df.groupBy('beds').count().show()\n",
    "#由考生填写\n",
    "df.describe(['baths','beds','price','sqft']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1e41ee4d",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.mllib\n",
    "import pyspark.mllib.regression\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "from pyspark.sql.functions import *\n",
    "df = df.select('price','baths','beds','sqft')\n",
    "df = df[df.baths > 0]\n",
    "df = df[df.beds > 0]\n",
    "df = df[df.sqft >0]\n",
    "df.describe(['baths','beds','price','sqft']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4b9cf14d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 这段代码多余的\n",
    "temp = df.rdd.map(lambda line:LabeledPoint(line[0],[line[1:]]))\n",
    "temp.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "02307f22",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.util import MLUtils\n",
    "from pyspark.mllib.linalg import Vectors\n",
    "from pyspark.mllib.feature import StandardScaler\n",
    "features = df.rdd.map(lambda row: row[1:])\n",
    "features.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cecbd910",
   "metadata": {},
   "outputs": [],
   "source": [
    "standardizer = StandardScaler()\n",
    "model = standardizer.fit(features)\n",
    "features_transform = model.transform(features)\n",
    "features_transform.take(5)\n",
    "lab = df.rdd.map(lambda row: row[0])\n",
    "lab.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ba395544",
   "metadata": {},
   "source": [
    "4.将标签lab和特征features transform进行zip操作。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a0efb82",
   "metadata": {},
   "outputs": [],
   "source": [
    "#由考生填写 \n",
    "transformedData = lab.zip(features_transform)\n",
    "#由考生填写 \n",
    "transformedData.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "835f0698",
   "metadata": {},
   "source": [
    "5.将transformedData转换为LabeledPoint类型,其中row[0]作为标签,row[1]作为特征向量。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a379a0f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "#由考生填写\n",
    "transformedData = transformedData.map(lambda row: LabeledPoint(row[0],[row[1]]))\n",
    "#由考生填写\n",
    "transformedData.take(5)\n",
    "traingingData,testingData = transformedData.randomSplit([.8,.2],seed=1234)\n",
    "from pyspark.mllib.regression import LinearRegressionWithSGD\n",
    "linearModel = LinearRegressionWithSGD.train(traingingData,1000,.2)\n",
    "linearModel.weights\n",
    "testingData.take(10)\n",
    "linearModel.predict([1.49297445326,3.52055958053,1.73535287287])"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark-2.4.5",
   "language": "python",
   "name": "pyspark-2.4.5"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
